package com.sh.data.engine.domain.datadev.flink.model.domain;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.sh.data.engine.common.enumDefinition.DSType;
import com.sh.data.engine.repository.dao.datadev.entity.FlinkNodeConfigEntity;
import lombok.Data;
import org.apache.commons.lang.StringUtils;

import java.util.Date;
import java.util.List;

@Data
public class FlinkSqlTaskDomain {
    /**
     * 主键
     */
    private Long id;

    /**
     * 项目id
     */
    private Long projectId;

    /**
     * 节点id
     */
    private Long nodeId;

    /**
     * 发布状态 0-未发布 1-发布
     */
    private Integer isPublished;

    /**
     * 创建人id
     */
    private String userId;

    /**
     * 修改人id
     */
    private String modifyUserId;

    /**
     * 发布人id
     */
    private String publishUserId;

    /**
     * 发布时间
     */
    private Date publishTime;

    /**
     * 描述
     */
    private String remark;

    /**
     * 是否是激活状态 0-否 1-是
     */
    private Integer isActive;

    /**
     * 创建时间
     */
    private Date createTime;

    /**
     * 更新时间
     */
    private Date updateTime;

    /**
     * 脚本内容
     */
    private String scriptContent;

    /**
     * 源表配置
     */
    private String sourceConfig;

    /**
     * 结果表配置
     */
    private String sinkConfig;

    /**
     * 维度表配置
     */
    private String dimConfig;

    /**
     * 环境参数配置
     */
    private String envConfig;

    /**
     * 调度配置
     */
    private String scheduleConfig;

    private Date nextExecTime;

    private Date nextStopTime;

    public List<FlinkSourceConfigDomain> getSourceConfigDomainList() {
        if (StringUtils.isBlank(sourceConfig)) {
            return Lists.newArrayList();
        }

        List<FlinkSourceConfigDomain> sourceConfigDomainList = Lists.newLinkedList();

        final JSONArray jsonArray = JSONArray.parseArray(sourceConfig);
        for (Object o : jsonArray) {
            JSONObject obj = (JSONObject) o;
            final String dsType = obj.getString("dsType");
            final DSType dsTypeEnum = DSType.valueOf(dsType);
            switch (dsTypeEnum) {
                case Kafka:
                    final FlinkKafkaSourceConfigDomain flinkKafkaSourceConfigDomain =
                        obj.toJavaObject(FlinkKafkaSourceConfigDomain.class);
                    sourceConfigDomainList.add(flinkKafkaSourceConfigDomain);
                    break;
            }
        }

        return sourceConfigDomainList;
    }

    public List<FlinkSinkConfigDomain> getSinkConfigDomainList() {
        if (StringUtils.isBlank(sinkConfig)) {
            return Lists.newArrayList();
        }

        List<FlinkSinkConfigDomain> sinkConfigDomainList = Lists.newLinkedList();

        final JSONArray jsonArray = JSONArray.parseArray(sinkConfig);
        for (Object o : jsonArray) {
            JSONObject obj = (JSONObject) o;
            final String dsType = obj.getString("dsType");
            final DSType dsTypeEnum = DSType.valueOf(dsType);
            switch (dsTypeEnum) {
                case Kafka:
                    final FlinkKafkaSinkConfigDomain flinkKafkaSinkConfigDomain =
                        obj.toJavaObject(FlinkKafkaSinkConfigDomain.class);
                    sinkConfigDomainList.add(flinkKafkaSinkConfigDomain);
                    break;
                case MySQL:
                case Oracle:
                    final FlinkJDBCSinkConfigDomain flinkJDBCSinkConfigDomain =
                        obj.toJavaObject(FlinkJDBCSinkConfigDomain.class);
                    sinkConfigDomainList.add(flinkJDBCSinkConfigDomain);
                    break;
                case HBase:
                    final FlinkHbaseSinkConfigDomain flinkHbaseSinkConfigDomain =
                        obj.toJavaObject(FlinkHbaseSinkConfigDomain.class);
                    sinkConfigDomainList.add(flinkHbaseSinkConfigDomain);
                    break;
                case Hive:
                    final FlinkHiveSinkConfigDomain flinkHiveSinkConfigDomain =
                        obj.toJavaObject(FlinkHiveSinkConfigDomain.class);
                    sinkConfigDomainList.add(flinkHiveSinkConfigDomain);
                    break;
            }
        }

        return sinkConfigDomainList;
    }

    public List<FlinkDimConfigDomain> getDimConfigDomainList() {
        if (StringUtils.isBlank(dimConfig)) {
            return Lists.newArrayList();
        }

        List<FlinkDimConfigDomain> dimConfigDomainList = Lists.newLinkedList();

        final JSONArray jsonArray = JSONArray.parseArray(dimConfig);
        for (Object o : jsonArray) {
            JSONObject jsonObject = (JSONObject) o;
            final String dsType = jsonObject.getString("dsType");
            final DSType dsTypeEnum = DSType.valueOf(dsType);
            switch (dsTypeEnum) {
                case MySQL:
                case Oracle:
                    dimConfigDomainList.add(jsonObject.toJavaObject(FlinkJDBCDimConfigDomain.class));
                    break;
                case Hive:
                    dimConfigDomainList.add(jsonObject.toJavaObject(FlinkHiveDimConfigDomain.class));
            }
        }

        return dimConfigDomainList;
    }

    public FlinkScheduleConfigDomain getScheduleConfigDomain() {
        if (StringUtils.isBlank(scheduleConfig)) {
            return null;
        }
        return JSONObject.parseObject(scheduleConfig, FlinkScheduleConfigDomain.class);
    }

    public void init(FlinkNodeConfigEntity entity) {
        this.id = entity.getId();
        this.projectId = entity.getProjectId();
        this.nodeId = entity.getNodeId();
        this.isPublished = entity.getIsPublished();
        this.userId = entity.getCreatorId();
        this.modifyUserId = entity.getUpdaterId();
        this.publishTime = entity.getPublishTime();
        this.remark = entity.getRemark();
        this.isActive = entity.getIsActive();
        this.createTime = entity.getCreateTime();
        this.updateTime = entity.getUpdateTime();
        this.scriptContent = entity.getSqlTaskConfig().getScriptContent();
        //    this.sourceConfig= JSON.toJSONString(entity.getSqlTaskConfig().getSourceConfig());
        //    this.sinkConfig=JSON.toJSONString(entity.getSqlTaskConfig().getSinkConfig());
        //    this.dimConfig=JSON.toJSONString(entity.getSqlTaskConfig().getDimConfig());
        this.sourceConfig = entity.getSqlTaskConfig().getSourceConfig();
        this.sinkConfig = entity.getSqlTaskConfig().getSinkConfig();
        this.dimConfig = entity.getSqlTaskConfig().getDimConfig();
        this.envConfig = entity.getEnvConfig();
        this.scheduleConfig = JSON.toJSONString(entity.getScheduleConfig());
        this.nextExecTime = entity.getNextExecTime();
        this.nextStopTime = entity.getNextStopTime();
        this.publishUserId = entity.getPublisherId();
    }
}
